package org.test.rocketmq.producer;

import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.time.StopWatch;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;
import org.test.rocketmq.RocketmqConfig;

@Component
public class RocketmqProducer implements SmartLifecycle {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketmqProducer.class);

    private final RocketmqConfig rocketmqConfig;

    private boolean running;
    public RocketmqProducer(RocketmqConfig rocketmqConfig) {
        this.rocketmqConfig = rocketmqConfig;
    }

    @Override
    public void start() {
        //生产者的组名
        DefaultMQProducer producer = new DefaultMQProducer(rocketmqConfig.getProducer().getGroup());
        //指定NameServer地址，多个地址以 ; 隔开
        producer.setNamesrvAddr(rocketmqConfig.getNamesrvAddr());
        try {
            /*
             * Producer对象在使用之前必须要调用start初始化，初始化一次即可
             * 注意：切记不可以在每次发送消息时，都调用start方法
             */
            producer.start();
            this.running = true;
            StopWatch stop = new StopWatch();
            stop.start();

            for (int i = 0; i < 10000; i++) {
                //创建一个消息实例，包含 topic、tag 和 消息体
                //如下：topic 为 "TopicTest"，tag 为 "push"
                Message message = new Message("TopicTest", "push", ("发送消息----Test" + i + "-----").getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult result = producer.send(message);
                LOGGER.info("发送响应：MsgId:{}，发送状态：{}", result.getMsgId(), result.getSendStatus());
            }
            stop.stop();
            LOGGER.info("----------------发送一万条消息耗时：{}s", stop.getTime(TimeUnit.SECONDS));
        } catch (Exception e) {
            LOGGER.error("send message exception", e);
        } finally {
            producer.shutdown();
        }
    }

    @Override
    public void stop() {
        // do nothing
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}
